package org.logstashplugins;

import java.lang.ref.SoftReference;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import co.elastic.logstash.api.Configuration;
import co.elastic.logstash.api.Context;
import co.elastic.logstash.api.Event;
import co.elastic.logstash.api.Filter;
import co.elastic.logstash.api.FilterMatchListener;
import co.elastic.logstash.api.LogstashPlugin;
import co.elastic.logstash.api.PluginConfigSpec;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPoolConfig;

@LogstashPlugin(name = "java_filter_redis")
public class JavaFilterRedis implements Filter {

	private static final Logger log = LoggerFactory.getLogger(JavaFilterRedis.class);
	public static PluginConfigSpec<String> HOST = PluginConfigSpec.stringSetting("host", "127.0.0.1");
	public static PluginConfigSpec<Long> PORT = PluginConfigSpec.numSetting("port", 6379);
	public static PluginConfigSpec<String> PASS = PluginConfigSpec.stringSetting("password", "");
	public static PluginConfigSpec<String> KEY_TYPE = PluginConfigSpec.stringSetting("key_type",
			KeyType.STRING.getType());
	// public static PluginConfigSpec<List<Object>> KEY_FIELD =
	// PluginConfigSpec.arraySetting("key_field");
	public static final PluginConfigSpec<String> SOURCE_FIELD = PluginConfigSpec.stringSetting("source", "message");
	public static final PluginConfigSpec<Map<String, Object>> FROM_TO_TARGET_FIELD = PluginConfigSpec
			.hashSetting("from_to_target_field");
	// 是否启用堆缓存
	public static final PluginConfigSpec<Boolean> USE_HEAP_CACHE = PluginConfigSpec.booleanSetting("use_heap_cache",
			false);

	// redis最大最小空闲连接数
	public static PluginConfigSpec<Long> MAXIDLE = PluginConfigSpec.numSetting("max_idle", 20);
	public static PluginConfigSpec<Long> MINIDLE = PluginConfigSpec.numSetting("min_idle", 8);
	public static PluginConfigSpec<Long> MAXTOTAL = PluginConfigSpec.numSetting("max_total", 200);

	public static volatile RedisUtil redisUtil = null;

	public static PluginConfigSpec<Long> INIT_KEYS_COUNT = PluginConfigSpec.numSetting("init_keys_count", 16 * 10000);
	
	
	// 为区号开启缓存
	public static final PluginConfigSpec<Boolean> USE_HEAP_CACHE_FOR_AREACODE = PluginConfigSpec.booleanSetting("use_heap_cache_for_areacode",
			false);
	
	

	@Override
	public Collection<PluginConfigSpec<?>> configSchema() {
		ArrayList<PluginConfigSpec<?>> arrayList = new ArrayList<>();
		arrayList.add(HOST);
		arrayList.add(PORT);
		arrayList.add(KEY_TYPE);
		arrayList.add(PASS);
		arrayList.add(SOURCE_FIELD);
//		arrayList.add(KEY_FIELD);

		arrayList.add(MAXIDLE);
		arrayList.add(MINIDLE);
		arrayList.add(MAXTOTAL);
		arrayList.add(INIT_KEYS_COUNT);

		arrayList.add(FROM_TO_TARGET_FIELD);

		arrayList.add(USE_HEAP_CACHE);
		
		arrayList.add(USE_HEAP_CACHE_FOR_AREACODE);

		return arrayList;
	}

	private String id;
	private String sourceField;
	private Map<String, String> fromToTargetField = new HashMap<String, String>();
	private static int initKeysCount;
	private boolean useHeapCache;
	private boolean useHeapCacheForAreacode;

	public static volatile SoftReference<Map<String, Map<String, String>>> HASH_CACHE = new SoftReference<Map<String, Map<String, String>>>(
			new ConcurrentHashMap<>(initKeysCount));
	
	public static volatile Map<String,Map<String,String>> NUMBER_MAP = new ConcurrentHashMap<String, Map<String,String>>(32 * 10000);
	public static volatile Map<String,Map<String,String>> AREACODE_MAP = new ConcurrentHashMap<String, Map<String,String>>(600);

	public JavaFilterRedis(String id, Configuration config, Context context) {
		super();
		System.out.println("初始化------------------》");
		this.id = id;
		this.sourceField = config.get(SOURCE_FIELD);

		System.out.println(sourceField);
		Map<String, Object> map = config.get(FROM_TO_TARGET_FIELD);
		if (map == null || map.isEmpty()) {
			throw new IllegalArgumentException("请传入目标字段。from_to_target_field（hash结构）字段不存在！");
		}
		map.keySet().stream().forEach(f -> {
			Object tmp = map.get(f);
			String t = String.valueOf(tmp);
			fromToTargetField.put(f, t);
		});

		initKeysCount = config.get(INIT_KEYS_COUNT).intValue();

		// 初始化Redis
		JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
		// 最大空闲数
		jedisPoolConfig.setMaxIdle(config.get(MAXIDLE).intValue());
		// 最大连接数
		jedisPoolConfig.setMaxTotal(config.get(MAXTOTAL).intValue());
		// 最小连接数
		jedisPoolConfig.setMinIdle(config.get(MINIDLE).intValue());
		jedisPoolConfig.setTestOnBorrow(true);
		jedisPoolConfig.setTestOnReturn(true);

		redisUtil = new RedisUtil(config.get(HOST), config.get(PORT).intValue(), config.get(PASS), jedisPoolConfig);

		// 获取连接测试
		Jedis conn = redisUtil.getConn();
		if (null == conn) {
			throw new RuntimeException("redis连接失败！！！，连接参数HOST[" + config.get(HOST) + "],PORT[" + config.get(PORT)
					+ "],PASS[" + config.get(PASS) + "]");
		} else {
			// 归还连接
			conn.close();
		}
		useHeapCache = config.get(USE_HEAP_CACHE);
		useHeapCacheForAreacode = config.get(USE_HEAP_CACHE_FOR_AREACODE);
		System.out.println("redis过滤插件初始化完毕！初始化信息：" + toString());

	}

	@Override
	public String getId() {

		return id;
	}

	@Override
	public Collection<Event> filter(Collection<Event> arg0, FilterMatchListener arg1) {

		arg0.stream().forEach(e -> {
			Object key = e.getField(sourceField);
			if (key != null) {
				String tmp = String.valueOf(key);
				Map<String, String> map = useHeapCacheForAreacode ? specifyForAreacode(tmp): specifyForNumber(tmp); // useHeapCache ? getKeyHash(tmp) : getKeyHashFromRedis(tmp);
				if (null != map) {
					fromToTargetField.keySet().stream().forEach(f -> {
						// 添加新的字段及字段对应的属性
						log.debug("添加新的字段及属性KEY[{}],原始字段[{}]，目标字段[{} ],目标字段值[{}]", tmp, f, fromToTargetField.get(f),
								map.get(f));
						e.setField(fromToTargetField.get(f), map.get(f));
					});
				}
			}
		});
		return arg0;
	}
	
	public Map<String, String> specifyForNumber(String key){
		Map<String, String> map = NUMBER_MAP.get(key);
		if(null != map) {
			return map;
		}
		// redis读取数据
		Map<String, String> all = getKeyHashFromRedis(key);
		if(null != all) {
			// 缓存
			NUMBER_MAP.put(key, all);
			return all;
		}
		return null;
	}
	
	// 区号
	public Map<String, String> specifyForAreacode(String key){
		Map<String, String> map = AREACODE_MAP.get(key);
		if(null != map) {
			return map;
		}
		// redis读取数据
		Map<String, String> all = getKeyHashFromRedis(key);
		if(null != all) {
			// 缓存
			AREACODE_MAP.put(key, all);
			return all;
		}
		return null;
	}

	public Map<String, String> getKeyHash(String key) {
		Map<String, Map<String, String>> cacheKey = HASH_CACHE.get();
		if (null == cacheKey) {
			synchronized (JavaFilterRedis.class) {
				if (null == cacheKey) {
					HASH_CACHE = new SoftReference<Map<String, Map<String, String>>>(
							new ConcurrentHashMap<>(initKeysCount));
					cacheKey = HASH_CACHE.get();
				}
			}
		}
		Map<String, String> map = cacheKey.get(key);
		if (null != map) {
			log.debug("缓存--->有数据，直接返回，KEY[{}],value[{}]", key, map);
			// System.out.println("缓存有数据，直接返回，KEY[" + key + "],value[" + map + "]");
			return map;
		}
		// 从redis获取数据
		Map<String, String> all = redisUtil.hgetAll(key);
		if (null != all && !all.isEmpty()) {
//			System.out.println("从Redis获取数据成功,KEY[" + key + "],VALUE[" + all + "]");
			log.debug("Redis获取数据成功,KEY[{}],VALUE[{}]", key, all);
			// 先缓存，后返回
			cacheKey.put(key, all);
			return all;
		}
		log.debug("REDIS未查询到数据ERROR，KEY[{}]", key);
		// System.out.println("REDIS未查询到数据，KEY[" + key + "]");
		return null;
	}

	// 是否使用对缓存
	public Map<String, String> getKeyHashFromRedis(String key) {
		// 从redis获取数据
		Map<String, String> all = redisUtil.hgetAll(key);
		if (null != all && !all.isEmpty()) {
			log.debug("-->Redis获取数据成功,KEY[{}],VALUE[{}]", key, all);
			// 先缓存，后返回
			return all;
		}
		log.debug("-->REDIS未查询到数据ERROR，KEY[{}]", key);
		// System.out.println("REDIS未查询到数据，KEY[" + key + "]");
		return null;
	}

	@Override
	public String toString() {
		StringBuilder builder = new StringBuilder();
		builder.append("JavaFilterRedis [id=");
		builder.append(id);
		builder.append(", sourceField=");
		builder.append(sourceField);
		builder.append(", fromToTargetField=");
		builder.append(fromToTargetField);
		builder.append("]");
		return builder.toString();
	}

}
